package cn._51doit.flink.day05;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * assignTimestampsAndWatermarks方法也是一个Transformation，不会改变数据的样式，仅仅会提取数据中的EventTime，然后生成WaterMark，向下游发送
 * assignTimestampsAndWatermarks方法返回的DataStream的并行度与调用该方法的DataStream并行度一致
 *
 *
 */
public class EventTimeTumblingWindowDemo3 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //1000,spark,4
        //2000,hive,5
        //4000,hive,2
        //4998,spark,1
        //4999,spark,2
        //6666,flink,2
        //7777,spark,3
        //8888,spark,1
        //9998,spark,2
        //10000,spark,200
        //14999,spark,100
        DataStreamSource<String> lines = env.socketTextStream("localhost", 9999);

        //(1000,spark,4)
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> tpStream = lines.map(new MapFunction<String, Tuple3<Long, String, Integer>>() {
            @Override
            public Tuple3<Long, String, Integer> map(String line) throws Exception {
                String[] fields = line.split(",");
                long timestamp = Long.parseLong(fields[0]);
                String word = fields[1];
                int count = Integer.parseInt(fields[2]);
                return Tuple3.of(timestamp, word, count);
            }
        });

        //在调用assignTimestampsAndWatermarks
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> tpStreamWithWaterMark = tpStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Long, String, Integer>>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(Tuple3<Long, String, Integer> tp) {
                return tp.f0;
            }
        });


        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = tpStreamWithWaterMark.map(new MapFunction<Tuple3<Long, String, Integer>, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(Tuple3<Long, String, Integer> tp) throws Exception {
                return Tuple2.of(tp.f1, tp.f2);
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = wordAndCount.keyBy(t -> t.f0);
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(5)));
        SingleOutputStreamOperator<Tuple2<String, Integer>> res = windowedStream.sum(1);
        res.print();
        env.execute();


    }
}
